package flinkstudy.application;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;


import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

public class FlinkReceiver {
    public static void main(String[] args) {


        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.enableCheckpointing(5000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");

        properties.setProperty("zookeeper.connect", "localhost:2181");
        properties.setProperty("group.id", "test");

        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");//设置日期格式
        FlinkKafkaConsumer<String> myConsumer =
                new FlinkKafkaConsumer<>("temp", new SimpleStringSchema(),properties);

        myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter());

        DataStream<Tuple2<String, Double>> keyedStream = env.addSource(myConsumer).

                 flatMap(new Splitter()).keyBy(0)
                .timeWindow(Time.seconds(5))
                .apply(new WindowFunction<Tuple2<String, Double>, Tuple2<String, Double>, Tuple, TimeWindow>() {

                   @Override
                    public void apply(Tuple key, TimeWindow window, Iterable<Tuple2<String, Double>> input,
                                      Collector<Tuple2<String, Double>> out) throws Exception {

                      //  System.out.println("WindowFunction apply");
                        double sum = 0L;
                        int count = 0;
                        for (Tuple2<String, Double> record : input) {
                            sum += record.f1;
                            count++;
                        }

                        Tuple2<String, Double> result = input.iterator().next();
                        result.f1 = (sum/count);
                        //System.out.println(df.format(new Date()));
                       // System.out.println("count:"+count);
                       // System.out.println("sum:"+sum);
                       // System.out.println("key:"+result.f0);
                      //  System.out.println("average:"+ result.f1);
                        out.collect(result);
                    }
                });



        keyedStream.print();

        try {
            // execute program
            env.execute("Flink Streaming Java API Skeleton");
        }catch (Exception ex)
        {
            ex.printStackTrace();
        }

    }
}
